package com.study.service.hystrix;

import com.alibaba.fastjson.JSON;
import com.netflix.hystrix.HystrixCommandGroupKey;
import com.netflix.hystrix.HystrixCommandKey;
import com.netflix.hystrix.HystrixObservableCommand;
import com.netflix.hystrix.strategy.properties.HystrixPropertiesCommandDefault;
import com.study.service.NacosClientService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.client.ServiceInstance;
import rx.Observable;
import rx.Subscriber;

import java.util.Collections;
import java.util.List;

/**
 * hystrixCommand 隔离策略 基于信号量实现
 */
@Slf4j
public class NacosClientHystrixObservableCommand extends HystrixObservableCommand<List<ServiceInstance>> {

    private final NacosClientService nacosClientService;


    private final List<String> serviceId;

    public NacosClientHystrixObservableCommand(NacosClientService nacosClientService,
                                               List<String> serviceId) {

        super(
                HystrixObservableCommand.Setter
                    .withGroupKey(HystrixCommandGroupKey
                            .Factory.asKey("NacosClientService"))
                    .andCommandKey(HystrixCommandKey
                            .Factory.asKey("NacosClientHystrixObservableCommand"))
                    .andCommandPropertiesDefaults(
                            HystrixPropertiesCommandDefault.Setter()
                            .withFallbackEnabled(true)//开启降级
                            .withCircuitBreakerEnabled(true) //开启熔断
                    )
                );
        this.nacosClientService =nacosClientService;
        this.serviceId =serviceId;

    }

    //要保护的方法
    @Override
    protected Observable construct() {
        return Observable.create(new Observable.OnSubscribe<List<ServiceInstance>>() {

            @Override
            public void call(Subscriber<? super List<ServiceInstance>> subscriber) {

                try {
                    if(!subscriber.isUnsubscribed()){
                        log.info("subcriber command task: [{}]、[{}]",
                                JSON.toJSONString(serviceId)
                                ,Thread.currentThread().getName());

                        serviceId.forEach(s -> {
                            subscriber.onNext(
                                    nacosClientService.getNacosClientInfo(s)
                            );
                        });
                        subscriber.onCompleted();
                        log.info("subcriber command complete: [{}]、[{}]",
                                JSON.toJSONString(serviceId)
                                ,Thread.currentThread().getName());
                    }

                }catch (Exception e){
                    subscriber.onError(e);
                }

            }
        });
    }

    //服务器降级
    @Override
    protected Observable<List<ServiceInstance>> resumeWithFallback() {
        return Observable.create(new Observable.OnSubscribe<List<ServiceInstance>>() {

            @Override
            public void call(Subscriber<? super List<ServiceInstance>> subscriber) {

                try {
                    if(!subscriber.isUnsubscribed()){
                        log.info("(Fallback)subcriber command task: [{}]、[{}]",
                                JSON.toJSONString(serviceId)
                                ,Thread.currentThread().getName());

                        subscriber.onNext(Collections.emptyList());
                        subscriber.onCompleted();
                        log.info("(Fallback)subcriber command complete: [{}]、[{}]",
                                JSON.toJSONString(serviceId)
                                ,Thread.currentThread().getName());
                    }

                }catch (Exception e){
                    subscriber.onError(e);
                }

            }
        });
    }
}
